package com.lnj.community.service.Impl;

import com.lnj.community.service.DataService;
import com.lnj.community.utiles.RedisKeyUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;

@Service
public class DataServiceImpl implements DataService {
    @Autowired
    private RedisTemplate redisTemplate;

    private SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
    /**
     * 将指定UV统计结果保存到Redis
     *
     * @param ip 客户端IP地址
     */
    @Override
    public void recordUV(String ip) {
        String redisKey = RedisKeyUtil.getUVKey(sdf.format(new Date()));
        //将IP地址添加到Redis的HyperLogLog数据结构中 表示该IP地址的访问记录
        redisTemplate.opsForHyperLogLog().add(redisKey,ip);
    }

    /**
     * 将指定DAU统计结果保存到Redis
     *
     * @param userId 用户ID
     */
    @Override
    public void recordDAU(int userId) {
        String redisKey = RedisKeyUtil.getDAUKey(sdf.format(new Date()));
        //以用户ID为偏移量，将该用户的登录状态设置为1，表示已登录  15号 ：1000100 表示用户1和用户5登录了
        redisTemplate.opsForValue().setBit(redisKey,userId,true);
    }

    /**
     * 统计指定日期范围内的UV 访问用户的记录数
     *
     * @param start 起始日期
     * @param end   结束日期
     * @return 统计结果
     */
    @Override
    public long calculateUV(Date start, Date end) {
        if(start == null || end == null){
            throw new IllegalArgumentException("参数不能为空");
        }
        //整理该日期范围内的key
        List<String> keyList = new ArrayList<>();
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(start);
        while(!calendar.getTime().after(end)){
            String key = RedisKeyUtil.getUVKey(sdf.format(calendar.getTime()));
            keyList.add(key);
            calendar.add(Calendar.DATE,1);
        }
        //合并这些数据
        String redisKey = RedisKeyUtil.getUVKey(sdf.format(start),sdf.format(end));
       redisTemplate.opsForHyperLogLog().union(redisKey, keyList.toArray());
        return redisTemplate.opsForHyperLogLog().size(redisKey);
    }

    /**
     * 统计指定日期范围内的DAU 活跃用户
     *
     * @param start 起始日期
     * @param end  结束日期
     * @return 统计结果
     */
    @Override
    public long calculateDAU(Date start, Date end) {
        if(start == null || end == null){
            throw new IllegalArgumentException("参数不能为空");
        }
        //整理该日期范围内的key   因为是位图 所以要转换为字节数组
        List<byte[]> keyList = new ArrayList<>();
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(start);
        while(!calendar.getTime().after(end)){
            //获取该日期的key
            String key = RedisKeyUtil.getDAUKey(sdf.format(calendar.getTime()));
            keyList.add(key.getBytes());
            calendar.add(Calendar.DATE,1);
        }
        //进行OR运算
        return (long) redisTemplate.execute(new RedisCallback() {
            @Override
            public Object doInRedis(RedisConnection con) throws DataAccessException {
                //这里是将多个key合并成一个key  因为是位图 所以要转换为字节数组
                String redisKey = RedisKeyUtil.getDAUKey(sdf.format(start),sdf.format(end));
                //这里是对于多个key的值进行OR运算
                con.bitOp(RedisStringCommands.BitOperation.OR,redisKey.getBytes(),keyList.toArray(new byte[0][0]));
                //统计这个位图中1的数量
                return con.bitCount(redisKey.getBytes());
            }
        });
    }
}
